package spark.core.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object rddTransOperators_Test {
  def main(args: Array[String]): Unit = {
    // 初始化Spark
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("rddTransOperators_Test")
    val sparkContext: SparkContext = new SparkContext(sparkConf)
    // 具体的业务操作
    // 获取源数据
    val fileRDD: RDD[String] = sparkContext.textFile("Input/agent.log")
    // 剔除无效数据并做格式转换
    val mapRDD: RDD[((String, String), Int)] = fileRDD.map(
      line => {
        val strings: Array[String] = line.split(" ")
        ((strings(1), strings(4)), 1)
      }
    )
    // 分组求和
    val groupRDD: RDD[((String, String), Iterable[Int])] = mapRDD.groupByKey()
    val mapValuesRDd: RDD[((String, String), Int)] = groupRDD.mapValues(
      data => data.sum
    )
    // 再次转换数据格式
    val newMapRDD: RDD[(String, (Int, Int))] = mapValuesRDd.map {
      case (tuple, i) => (tuple._1, (tuple._2.toInt, i))
    }
    // 再次分组 排序 取部分数据
    val groupByKeyRDD: RDD[(String, Iterable[(Int, Int)])] = newMapRDD.groupByKey()
    val resultRDD: RDD[(String, List[(Int, Int)])] = groupByKeyRDD.mapValues(
      iter => {
        iter.toList.sortBy(data => data._2)(Ordering.Int.reverse).take(3)
      }
    )
    // 再次转换数据格式 去掉List
    val resultRDD1: RDD[(String, String)] = resultRDD.map {
      case (string, list) =>
        ("城市编号："+string, "（广告编号，点击次数）："+list.mkString)
    }

    resultRDD1.collect().foreach(println)

    // 关闭环境
    sparkContext.stop()
  }

}
